#
# If you don't know what the '{{{' '}}}' are for, they are used
#  for folds in the vim text editor (:help fold).
#

# {{{ Intro

=head1 NAME

Finance::YahooStreamer - Interface to Yahoo! Finance Streamer.

=head1 SYNOPSIS

 use Finance::YahooStreamer;

 my $streamer = Finance::YahooStreamer->new(
					symbols => [qw(YHOO GOOG MSFT)],
					);

 $streamer->receive;

=head1 DESCRIPTION

This library provides an interface that can be used to access 
Yahoo! Finance Streamer market data.

=cut

# }}}

package Finance::YahooStreamer;
use strict;
use warnings;

use Carp;

our $VERSION = '0.03';

use IO::Socket::INET;
use IO::Select;
use JSON;

my $AGENT  = __PACKAGE__ . "/$VERSION";
my $SERVER = 'streamerapi.finance.yahoo.com';
my $PORT   = 80;

# max size of buffer used for recv()
my $RECV_MAX = 1024;

# default timeout in seconds of socket
# Used for connect(), recv().
my $TIMEOUT = 60;

=head1 OPERATIONS

=cut

# {{{ new

=head2 Finance::YahooStreamer->new;

	Return: defined object on success, croak otherwise

The I<new> sub stores the values passed to it for use by other
subroutines later.  For example, if you wanted to use a
subroutine that required a value for I<symbols> to be defined,
you could do it like so.

  $obj = Finance::YahooStreamer->new(symbols => [qw($your_symbols)]) or die "error new()";

=cut

sub new
{
  my ($class, %args) = @_;

  bless {%args}, $class;
}

# }}}

# {{{ connect

=head2 $obj->connect;

	Returns: IO::Socket::INET object on success, croak on error

	Requires the following object attributes:
		symbols [agent] [timeout]

The I<connect> sub is used to initiate a connection with the data server.
You do B<not> need to call I<connect>, I<receive> will do it for you.

The object attributes I<symbols>, optional I<agent>, and an optional I<timeout>,
must be defined in the I<streamer> object before using this subroutine.
Each is describe in more detail below.

 $obj->{symbols} = [@symbols];
 $obj->{agent} => $agent;	# optional
 $obj->{timeout} => $timeout;	# optional

The I<agent> field determines the id of this library when it connects to
a Streamer server.  By default the id is the name of this library.  The
string should be one line with B<no> carriage return ('\n').

 $agent = "My Server 1.01";

The I<timeout> specifies the maximum number of seconds to wait for the
connection to succeed.  The default value of B<60 seconds> is used
if no value is specified.

 $timeout = 60;

 my $sock = $obj->connect or die "error connect()";

=cut

sub connect
{
  my ($self) = @_;

  my $symbols = $self->{symbols};
  my $timeout = $self->{timeout} || $TIMEOUT;
  my $agent   = $self->{agent} || $AGENT;

  unless ($symbols && scalar(@$symbols))
  {
    my $time = localtime(time);
    croak "$time: connect(), no symbols\n";
  }

  my $sock = IO::Socket::INET->new(
    PeerAddr => $SERVER,
    PeerPort => $PORT,
    Proto    => 'tcp',
    Timeout  => $timeout,
  );
  unless ($sock)
  {
    my $time = localtime(time);
    croak "$time: connect(), socket creation failed: $!\n";
  }

# GET /streamer/1.0?s=YHOO,USD=X&o=^DJI,^IXIC,^GSPC,^TYX,^RUT,YHOO,URE,UPRO,GE,ERX,EDC,MSFT,GOOG,QQQQ,LVS&k=b00,b60,c10,g00,h00,j10,l10,p20,t10,v00,z08,z09&j=c10,l10,p20,t10&r=0&marketid=us_market&callback=parent.yfs_u1f&mktmcb=parent.yfs_mktmcb&gencallback=parent.yfs_gencb
# GET /streamer/1.0?s=GOOG,USD=X&o=^DJI,^IXIC,^GSPC,^TYX,^RUT,GOOG,YHOO,IYC,USD=X,URE,UPRO,GE,ERX,EDC,MSFT&k=a00,a50,c10,g00,h00,j10,l10,p20,t10,v00,z08,z09&j=c10,l10,p20,t10&r=0&marketid=us_market&callback=parent.yfs_u1f&mktmcb=parent.yfs_mktmcb&gencallback=parent.yfs_gencb

# For the meaning of these keys, please see:
# http://blog.underdog-projects.net/tag/streaming-api/
# http://search.cpan.org/dist/Finance-YahooQuote/YahooQuote.pm#The_available_custom_fields

  my %QUERY = (
    s           => join(',', @$symbols),
    o           => join(',', @$symbols),
    k           => 'a00,a50,b00,b60,c10,g00,h00,j10,l10,p20,t10,v00,z08,z09',
    j           => 'c10,l10,p20,t10',
    r           => 0,
    marketid    => 'us_market',
    callback    => 'parent.yfs_u1f',
    mktmcb      => 'parent.yfs_mktmcb',
    gencallback => 'parent.yfs_gencb',
  );

  my $query;
  foreach my $k (keys %QUERY)
  {
    $query .= "&" if $query;
    $query .= $k . "=" . $QUERY{$k};
  }

  # message used for request
  my $msg =
      "GET /streamer/1.0?$query HTTP/1.1\r\n"
    . "Accept-Language: en\r\n"
    . "Connection: Keep-Alive\r\n"
    . "Host: $SERVER\r\n";

  $msg .= "User-Agent: $agent\r\n" if $agent;
  $msg .= "\r\n";

  warn $msg if $self->{'debug'};

  unless ($sock->send($msg))
  {
    my $time = localtime(time);
    croak "$time: connect(), initial send() failed: $!\n";
  }

  $self->{'sock'} = $sock;
  return $self->{'sock'};
}

# }}}

# {{{ receive

=head2 $obj->receive;

	Returns: does not return

	Requires the following object attributes:
 		sub_recv symbols [timeout] [sub_hrtbt]

The I<receive> subroutine deals with all the issues of connecting to the
server, receiving data, etc, and executes the subroutine specified by
I<sub_recv>, passing two arguments, the symbol and its data,
every time a quote is received.

The object attributes I<sub_recv>, I<symbols>,
optional I<timeout> and optional I<sub_hrtbt> must be defined
before using this subroutine.

 $obj->{sub_recv} = $sub_ref;
 $obj->{sub_hrtbt} = $sub_ref_heartbeat;

The I<sub_recv> value is a reference to a subroutine to be executed
when new quote data arrives.  Two arguments, the symbol and its data.

The values I<symbols>, and I<timeout> are used for the I<connect> subroutine.
See the section on I<connect> for more information.

The I<timeout> value, while it is used for I<connect>, is also used in 
this subroutine to specify the maximum number of seconds to wait for 
new data to arrive before reconnecting.  The default value of B<60 seconds> 
is used if no value is specified.

The I<sub_hrtbt> value is a reference to a subroutine to be executed
when a B<heartbeat> happens.  One argument, the data including B<unixtime>,
will be passed to this subroutine when executed.

Error messages may be displayed.  Messages about errors receiving data 
will indicate why and may result in a reconnection.  Messages about
the status indicated in the received data are for information purposes 
and do not usually result in a reconnect.  All messages are displayed
to STDERR and so can be easily redirected.  An example of how to turn off
messages is below, where "a.out" is the name of the program and "2" is the
number of the file descriptor representing standard error.

 a.out 2>/dev/null

=cut

sub receive
{
  my $self    = shift;
  my $sub     = $self->{sub_recv};
  my $hb_sub  = $self->{sub_hrtbt};
  my $timeout = $self->{timeout} || $TIMEOUT;

  while (1)    # connect loop
  {
    $self->connect() if !$self->{'sock'} or !$self->{'sock'}->connected();
    unless ($self->{'sock'}->connected())
    {
      my $time = localtime(time);
      croak "$time: receive(), connect() failed: $!\n";
    }

    my $sel = IO::Select->new;
    $sel->add($self->{'sock'});

    while (1)    # read loop
    {
      next if !$sel->can_read($timeout);

      my $buf;
      $self->{'sock'}->recv($buf, $RECV_MAX);
      unless ($buf)
      {
        my $time = localtime(time);
        croak "$time: receive(), err #1\n";
      }

      next if $buf !~ m{<script>};
      while ($buf =~ m{<script>(.*?)</script>}i)    # parse loop
      {
        my $script = $1;
        $buf = $';
        my $cb   = $1 if $script =~ m{parent\.(\w+)};
        my $json = $1 if $script =~ m{^.*?\((.*?)\).*$};
        $json =~ s{([a-z]\d{2})}{"$1"}g;
        my $obj = decode_json($json);
        if ($cb eq 'yfs_u1f')
        {
          my @symbols = keys %$obj;
          my $symbol  = shift @symbols;
          $sub->($symbol, $obj->{$symbol}) if defined($sub);
        }
        elsif ($cb eq 'yfs_mktmcb')
        {
          $hb_sub->($obj) if defined($hb_sub);
        }
        else
        {
          warn "receive(): unknown callback: $cb";
        }
      }
      last if $self->{'debug'};
    }
    close($self->{'sock'});
    last if $self->{'debug'};
  }
}

# }}}

1;

=head1 PREREQUISITES

 Module             Version
 ------             -------
 IO::Socket::INET   1.25
 IO::Select         1.14

=head1 AUTHOR

Yen-Ming Lee E<lt>leeym@leeym.comE<gt>

=head1 COPYRIGHT

Copyright (c) 2009, Yen-Ming Lee. All Rights Reserved.
This module is free software.  It may be used, redistributed
and/or modified under the same terms as Perl itself.

=cut

# vi:foldmethod=marker
